package com.hliushi.utils

import java.sql.DriverManager

import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.kafka010.OffsetRange

import scala.collection.mutable

/**
 * descriptions: 用来手动维护偏移量到MySQL中的工具类
 *
 * author: Hliushi
 * date: 2021/7/1 19:51
 */
object OffsetUtils {


  /**
   * 根据参数查询偏移量信息并封装成Map返回
   *
   * @param groupId 消费者组名称
   * @param topic   主题
   * @return 偏移量信息封装成的Map
   */
  def getOffsetMap(groupId: String, topic: String): Map[TopicPartition, Long] = {
    // 1.加载驱动并获取连接
    val conn = DriverManager.getConnection("jdbc:mysql://cvm-node01:3306/bigdata?characterEncoding=UTF-8&useSSL=false", "root", "123456")
    // 2.编写SQL语句

    /**
     * bug点, partition是MySQL中关键字, 所以需要使用 ``将字段包裹起来
     * 不然就会出现一个SQL语法错误 You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'partition, offset
     * from t_offset where groupid = 'SparkKafka' and topic = 'covi' at line 1
     */
    val sql: String =
      """
        |select  `partition`, `offset`
        |from t_offset
        |where groupid = ? and topic = ?
        |""".stripMargin

    // 3.创建预编译语句对象
    val preSate = conn.prepareStatement(sql)

    // 4.设置参数并执行
    preSate.setString(1, groupId)
    preSate.setString(2, topic)
    val resultSet = preSate.executeQuery()

    // 5.获取返回值并封装成Map
    val offsetsMap: mutable.Map[TopicPartition, Long] = mutable.Map[TopicPartition, Long]()
    while (resultSet.next()) {
      val partition = resultSet.getInt("partition")
      val offset = resultSet.getLong("offset")

      offsetsMap += new TopicPartition(topic, partition) -> offset
    }

    // 6.关闭资源
    resultSet.close()
    preSate.close()
    conn.close()

    // 7.返回Map
    offsetsMap.toMap
  }


  /**
   * 将消费者组的偏移量信息存入MySQL
   * #
   * CREATE TABLE `t_offset` (
   * `topic` varchar(255) NOT NULL,
   * `partition` int(11) NOT NULL,
   * `groupid` varchar(255) NOT NULL,
   * `offset` bigint(20) DEFAULT NULL,
   * PRIMARY KEY (`topic`,`partition`,`groupid`)
   * ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
   *
   * @param groupId 消费者组名称
   * @param offsets 偏移量信息
   */
  def saveOffsets(groupId: String, offsets: Array[OffsetRange]): Unit = {
    // 1.加载驱动并获取连接
    val conn = DriverManager.getConnection("jdbc:mysql://cvm-node01:3306/bigdata?characterEncoding=UTF-8&useSSL=false",
      "root",
      "123456")
    // 2.编写SQL语句
    val sql: String =
      """
        |replace into t_offset (`topic`, `partition`, `groupid`, `offset`)
        |value(?, ?, ?, ?)
        |""".stripMargin

    // 3.创建预编译语句对象
    val preSate = conn.prepareStatement(sql)

    // 4.设置参数并执行
    for (o <- offsets) {
      preSate.setString(1, o.topic)
      preSate.setInt(2, o.partition)
      preSate.setString(3, groupId)
      preSate.setLong(4, o.untilOffset) // 这里选择了 untilOffset, 没有选择fromOffset
      preSate.executeUpdate()
    }
    // 5.关闭资源
    preSate.close()
    conn.close()
  }
}